package top.malaoshi.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class Consumer {
    // 消费监听
//    @KafkaListener(topics = {"userlog2"})
//    public void onMessage(ConsumerRecord<?, ?> record) {
//        // 消费的哪个topic、partition的消息,打印出消息内容
//        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "-" + record.value());
//    }


//    @KafkaListener(topicPartitions = {
//            @TopicPartition(topic = "userlog2", partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "9"))
//    })
//    public void onMessage(ConsumerRecord<?, ?> record) {
//        System.out.println("消费：topic:"+record.topic()+"-partition:"+record.partition()+"-offset:"+record.offset()+"-value:"+record.value());
//    }

    // 消费监听
//    @KafkaListener(topics = {"userlog2"})
//    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
//        // 消费的哪个topic、partition的消息,打印出消息内容
//        System.out.println("简单消费：" + record.topic() + "-" + record.partition() + "-" + record.value());
//        ack.acknowledge();
//    }

    @KafkaListener(id = "consumer1",topics = {"userlog2"})
    public void onMessage(@Payload String message,
                          @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                          @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                          @Header(KafkaHeaders.OFFSET) String offset,
                          org.apache.kafka.clients.consumer.Consumer consumer) {

        // 消费的哪个topic、partition的消息,打印出消息内容
        System.out.println(String.format("消费：topic:%s-partition:%s-offset:%s-value:%s", topic,partition,offset,message));
        consumer.commitAsync();
    }
}
